package org.jxkj.app

import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{when, _}
import org.jxkj.data.IDaoProvider

import java.text.SimpleDateFormat
import java.util.Date

/**
 * 1、启停统计分析
 */
object StartStop {
  @Deprecated
  private def calculateData(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider, time: Array[String], step: Int): Unit = {
    var startTime = time(0)
    val firstTime = time(0)
    val endTime = time(1)
    val endTime2 = endTime + " 23:59:59"
    val sdf = new SimpleDateFormat("yyyy/MM/dd")
    val sdfs = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")

    //var startTimeStamp = sdf.parse(startTime).getTime()
    //var nextDate = sdf.format(new Date(startTimeStamp-3600*24*1000))
    var endTimeStamp = sdf.parse(endTime).getTime()
    var lastDate = sdf.format(new Date(endTimeStamp - 3600 * 24 * 1000))
    var lastDate2 = sdf.format(new Date(endTimeStamp - 60 * 15))


    //读mysql
    val jdbcDF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_ta_start_stop_point")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()
    //.persist(StorageLevel.MEMORY_AND_DISK_SER)
    //hms_eo_trfloss_config_DF.show()
    jdbcDF.createOrReplaceTempView("hms_ta_start_stop_point")

    import ss.implicits._
    val point_code: Array[String] = jdbcDF.map(x => x.getString(2)).collect()
    //val point_code= Array("JKXT_EJ_AI03079","JKXT_EJ_AI03081","JKXT_EJ_AI03239","JKXT_EJ_DI11543","JKXT_EJ_DI11542")
    //for(i <- 0 until  point_code.length){println(point_code(i))}
    //用到15分钟补点
    //GetData.get(ss,sc,hTable,point_code,time,step)
    GetData.getNoLoopFromWeb(ss, sc, hTable, point_code, time, step)
    //ss.sql("select * from t_ods_hbase_edos").show(500)

    //.filter($"checkPoint"=== "JKXT_EJ_DI11535" )
    val odsDF = ss.sql("select a.sp_id,a.group_id,checkPoint,createTime,pvalue,begin_state,end_state,top_level,low_level" +
      " from hms_ta_start_stop_point  a " +
      " join t_ods_hbase_edos b on b.checkPoint=a.point_code ")

    val win = Window.partitionBy("checkPoint").orderBy("createTime")
    val dfWithLag = odsDF.withColumn("lag_value", lag($"pvalue", 1).over(win))
      .withColumn("flag", when($"lag_value" === $"pvalue", 0).otherwise(1)).filter($"flag" === 1)
      .withColumn("lead_time", lead($"createTime", 1).over(win))
      .filter("pvalue=begin_state")

    //dfWithLag.orderBy("checkPoint","createTime").show(500)
    dfWithLag.createOrReplaceTempView("t_dftag")

    //读mysql
    val jdbcDF4 = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_ta_strat_stop_records")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()

    jdbcDF4.createOrReplaceTempView("hms_ta_strat_stop_records")
    //checkPoint point_code,
    val resultDf = ss.sql("select a.sp_id,a.group_id,createTime start_time,lead_time end_time,unix_timestamp(lead_time,'yyyy/MM/dd HH:mm:ss')-unix_timestamp(createTime,'yyyy/MM/dd HH:mm:ss') last_time" +
      ",case when unix_timestamp(lead_time,'yyyy/MM/dd HH:mm:ss')-unix_timestamp(createTime,'yyyy/MM/dd HH:mm:ss') between nvl(low_level,0) and nvl(top_level,1000) then 1 else 0 end state" +
      ",'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time  from t_dftag a " +
      "left join hms_ta_strat_stop_records b on a.sp_id=b.sp_id and a.group_id=b.group_id and a.createTime=b.start_time" +
      " where b.sp_id is null and a.lead_time is not null")

    val tablename = "hms_ta_strat_stop_records"
    val delString = s"delete from $tablename where substr(start_time,1,10) between '$startTime' and '$endTime'"
    MysqlDml.delete2(delString)

    //写mysql
    resultDf.write
      .format("jdbc")
      .mode("append")
      .option("url", MysqlDml.url2)
      .option("dbtable", tablename)
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .save()


  }

  /**
   * 启停分析20210114
   *
   * @param ss
   * @param sc
   * @param hTable
   * @param time
   * @param step
   */
  def calcStartStop(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider, time: Array[String], step: Int): Unit = {
    var startTime = time(0)
    val firstTime = time(0)
    val endTime = time(1)
    //    val endTime2=endTime+" 23:59:59"
    val sdf = new SimpleDateFormat("yyyy/MM/dd")
    val sdfs = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")

    //var startTimeStamp = sdf.parse(startTime).getTime()
    //var nextDate = sdf.format(new Date(startTimeStamp-3600*24*1000))
    //    var endTimeStamp = sdf.parse(endTime).getTime()
    //    var lastDate = sdf.format(new Date(endTimeStamp-3600*24*1000))
    //    var lastDate2 = sdf.format(new Date(endTimeStamp-60*15))


    //读mysql
    val jdbcDF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_cm_scene_point")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()
    //.persist(StorageLevel.MEMORY_AND_DISK_SER)
    //hms_eo_trfloss_config_DF.show()
    jdbcDF.createOrReplaceTempView("hms_cm_scene_point")

    val jdbcDF2 = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_cm_scene_item")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()
    //.persist(StorageLevel.MEMORY_AND_DISK_SER)
    //hms_eo_trfloss_config_DF.show()
    jdbcDF2.createOrReplaceTempView("hms_cm_scene_item")


    val jdbcDF3 = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_cm_scene")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()
    //.persist(StorageLevel.MEMORY_AND_DISK_SER)
    //hms_eo_trfloss_config_DF.show()
    jdbcDF3.createOrReplaceTempView("hms_cm_scene")

    val pointDf = ss.sql("select c.scene_id,c.item_id,c.point_code,a.scene_name " +
      "from hms_cm_scene a join hms_cm_scene_point c " +
      " on a.scene_id=c.scene_id " +
      " join hms_cm_point b on c.point_code=b.point_code and b.point_type='on-off'" +
      " where substr(a.scene_name,-2)='启停'")
    // and substr(c.point_name,-2)='启动'
    pointDf.createOrReplaceTempView("t_etl_point")

    import ss.implicits._
    val point_code: Array[String] = pointDf.map(x => x.getString(2)).collect()
    //健康评估月报告-检修排水泵启停
    //.filter($"scene_id"=== "2ec35816260d4bd8b59b6ef2ddab0a2e" )
    //val point_code= Array("JKXT_EJ_DI07600","JKXT_EJ_DI07601","JKXT_EJ_DI07602")
    for (i <- 0 until point_code.length) {
      println(point_code(i))
    }

    //GetData.getBySecond(ss,sc,hTable,point_code,time,step,1)
    //往前取一天解决跨天 机组下所有泵保证取值周期一致
    GetData.getNoLoopFromWeb(ss, sc, hTable, point_code, time, step)
    //ss.sql("select * from t_ods_hbase_edos").show(500)

    //.filter($"checkPoint"=== "JKXT_EJ_DI11535" )
    val odsDF = ss.sql("select a.scene_id,a.item_id,checkPoint,createTime,pvalue,1 begin_state" +
      ",case when scene_name='压油泵启停' then 2000 when scene_name in ('技供漏油泵启停','机坑漏油泵启停') then 10000 when scene_name  in ('顶盖泵启停','检修泵启停','渗漏泵启停','低压机启停','中压机启停')  then 10001 else 10001 end  end_state" +
      " from t_etl_point  a " +
      " join t_ods_hbase_edos_tmp b on b.checkPoint=a.point_code ")
    //odsDF.orderBy("item_id","createTime").show(500)

    odsDF.createOrReplaceTempView("t_odsDF")
    val win = Window.partitionBy("checkPoint").orderBy("createTime")
    val win2 = Window.partitionBy("item_id").orderBy("createTime", "lead_time")
    val dfWithLag = odsDF.withColumn("lag_value", lag($"pvalue", 1).over(win))
      .withColumn("flag", when($"lag_value" === $"pvalue", 0).otherwise(1)).filter($"flag" === 1)
      .withColumn("lead_time", lead($"createTime", 1).over(win))
      .withColumn("lag_time", lag($"createTime", 1).over(win))
      .filter("pvalue=begin_state")


    //dfWithLag.orderBy("checkPoint","createTime").show()
    val t_dftag = dfWithLag.withColumn("max_lead_time", max($"lead_time").over(win2)).cache()
    t_dftag.createOrReplaceTempView("t_dftag")
    println("---")
    val odsDF3 = ss.sql("select a.scene_id,a.item_id,checkPoint,createTime,lead_time,lag_time,max_lead_time" +
      ",lag(max_lead_time,1) over(partition by item_id order by createTime,lead_time) lag_end_time" +
      ",max(lag_time) over(partition by item_id order by createTime) max_lag_time" +
      " from t_dftag a"
    )
    //odsDF3.orderBy("item_id","createTime").show(500)
    odsDF3.createOrReplaceTempView("t_odsDF3")

    ss.sql("select a.*,sum(case when createTime<=lag_end_time then 0 else 1 end) over(partition by item_id order by createTime,lead_time rows between unbounded preceding and current row) as group_flag" +
      ",case when createTime<=lag_end_time then 0 else 1 end is_sum" +
      " from t_odsDF3 a").createOrReplaceTempView("t_etl")
    //ss.sql("select * from t_etl").show(500)
    ss.sql("select scene_id,item_id,group_flag" +
      ",min(createTime) start_time,max(lead_time) end_time" +
      ",unix_timestamp(max(lead_time),'yyyy/MM/dd HH:mm:ss')-unix_timestamp(min(createTime),'yyyy/MM/dd HH:mm:ss') start_to_end" +
      ",sum(unix_timestamp(lead_time,'yyyy/MM/dd HH:mm:ss')-unix_timestamp(createTime,'yyyy/MM/dd HH:mm:ss')) run_interval" +
      ",sum(case when createTime>nvl(lag_end_time,max_lag_time) then unix_timestamp(createTime,'yyyy/MM/dd HH:mm:ss')-unix_timestamp(nvl(lag_end_time,max_lag_time),'yyyy/MM/dd HH:mm:ss') else 0 end) stop_interval" +
      ",count(distinct checkPoint) device_cnt" +
      " from t_etl group by scene_id,item_id,group_flag").createOrReplaceTempView("t_etl2")

    //ss.sql("select *from t_etl2").show(500)

    ///////////////////////////////////////////////////////清洗掉单泵超时的数据，再计算整体
    val odsDF3_2 = ss.sql("select a.scene_id,a.item_id,checkPoint,createTime,lead_time,lag_time,max_lead_time" +
      ",lag(max_lead_time,1) over(partition by item_id order by createTime,lead_time) lag_end_time" +
      ",max(lag_time) over(partition by item_id order by createTime) max_lag_time" +
      " from t_dftag a " +
      " where unix_timestamp(lead_time,'yyyy/MM/dd HH:mm:ss')-unix_timestamp(createTime,'yyyy/MM/dd HH:mm:ss')<end_state"
    )
    //odsDF3.orderBy("item_id","createTime").show(500)
    odsDF3_2.createOrReplaceTempView("t_odsDF3_2")

    ss.sql("select a.*,sum(case when createTime<=lag_end_time then 0 else 1 end) over(partition by item_id order by createTime,lead_time rows between unbounded preceding and current row) as group_flag" +
      ",case when createTime<=lag_end_time then 0 else 1 end is_sum" +
      " from t_odsDF3_2 a").createOrReplaceTempView("t_etl_2")
    //ss.sql("select * from t_etl").show(500)
    ss.sql("select scene_id,item_id,group_flag" +
      ",min(createTime) start_time,max(lead_time) end_time" +
      ",unix_timestamp(max(lead_time),'yyyy/MM/dd HH:mm:ss')-unix_timestamp(min(createTime),'yyyy/MM/dd HH:mm:ss') start_to_end" +
      ",sum(unix_timestamp(lead_time,'yyyy/MM/dd HH:mm:ss')-unix_timestamp(createTime,'yyyy/MM/dd HH:mm:ss')) run_interval" +
      ",sum(case when createTime>nvl(lag_end_time,max_lag_time) then unix_timestamp(createTime,'yyyy/MM/dd HH:mm:ss')-unix_timestamp(nvl(lag_end_time,max_lag_time),'yyyy/MM/dd HH:mm:ss') else 0 end) stop_interval" +
      ",count(distinct checkPoint) device_cnt" +
      " from t_etl_2 group by scene_id,item_id,group_flag").createOrReplaceTempView("t_etl2_2")
    /////////////以上为添加清洗

    //持续时长、运行时长、间隔时长


    //读mysql
    val jdbcDF4 = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_ta_strat_stop_data")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()

    jdbcDF4.createOrReplaceTempView("hms_ta_strat_stop_data")
    //checkPoint point_code,
    val resultDf = ss.sql("select a.scene_id,a.item_id,checkPoint as point_code,createTime start_time,lead_time end_time" +
      ",unix_timestamp(lead_time,'yyyy/MM/dd HH:mm:ss')-unix_timestamp(createTime,'yyyy/MM/dd HH:mm:ss') last_time" +
      ",unix_timestamp(createTime,'yyyy/MM/dd HH:mm:ss')-unix_timestamp(lag_time,'yyyy/MM/dd HH:mm:ss') stop_interval" +
      //",'' state" +
      ",'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time  from t_dftag a " +
      "left join hms_ta_strat_stop_data b on a.checkPoint=b.point_code and a.item_id=b.item_id and a.createTime=b.start_time" +
      " where b.point_code is null and a.lead_time is not null")

    val tablename = "hms_ta_strat_stop_data"
    val delString = s"delete from $tablename where start_time between '$startTime' and '$endTime'"
    MysqlDml.delete2(delString)

    //写mysql
    resultDf.write
      .format("jdbc")
      .mode("append")
      .option("url", MysqlDml.url2)
      .option("dbtable", tablename)
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .save()
    //////////场景类型 机组统计
    //读mysql
    val jdbcDF5 = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_ta_strat_stop_item")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()

    jdbcDF5.createOrReplaceTempView("hms_ta_strat_stop_item")

    val jdbcDF6 = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_ta_strat_stop_item_clean")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()

    jdbcDF6.createOrReplaceTempView("hms_ta_strat_stop_item_clean")

    //checkPoint point_code,
    val resultDf2 = ss.sql("select a.scene_id,a.item_id,a.start_time,a.end_time,a.start_to_end,a.run_interval,a.stop_interval,a.device_cnt" +
      ",'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time  from t_etl2 a " +
      //修改a.start_time=b.start_time 为a.start_time>=b.start_time and a.end_time<=b.end_time
      "left join hms_ta_strat_stop_item b on a.scene_id=b.scene_id and a.item_id=b.item_id and a.start_time>=b.start_time and a.end_time<=b.end_time" +
      " where b.item_id is null and a.end_time is not null")
    // 在delete之前resultDf2.show()
    println("-------")
    val tablename2 = "hms_ta_strat_stop_item"
    val delString2 = s"delete from $tablename2 where start_time between '$startTime' and '$endTime'"
    MysqlDml.delete2(delString2)
    // 在delete后resultDf2.show()
    //写mysql
    resultDf2.write
      .format("jdbc")
      .mode("append")
      .option("url", MysqlDml.url2)
      .option("dbtable", tablename2)
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .save()

    val resultDf3 = ss.sql("select a.scene_id,a.item_id,a.start_time,a.end_time,a.start_to_end,a.run_interval,a.stop_interval,a.device_cnt" +
      ",'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time  from t_etl2_2 a " +
      "left join hms_ta_strat_stop_item_clean b on a.scene_id=b.scene_id and a.item_id=b.item_id and a.start_time>=b.start_time and a.end_time<=b.end_time" +
      " where b.item_id is null and a.end_time is not null")
    // 在delete之前resultDf2.show()
    println("-------")
    val tablename3 = "hms_ta_strat_stop_item_clean"
    val delString3 = s"delete from $tablename3 where start_time between '$startTime' and '$endTime'"
    MysqlDml.delete2(delString3)
    // 在delete后resultDf2.show()
    //写mysql
    resultDf3.write
      .format("jdbc")
      .mode("append")
      .option("url", MysqlDml.url2)
      .option("dbtable", tablename3)
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .save()


  }
}